#include <stdio.h>
#include <pthread.h>
#include <sys/eventfd.h>

#include <string>

#include "tcp_data_handler.h"
#include "tcp_acceptor.h"
#include "application.h"
#include "thread.h"
#include "circular_queue.h"

using namespace kingnet ;

class test_app ;
DECLARE_APPLICATION_INSTANCE(test_app) ;

class test_connection : public tcp_data_handler
{
public:
    test_connection()
    {
        m_last_time = time(0) ;
        m_timer.set_owner(this) ;
    }

/*
    int get_packet_info(const char* data,int size,packet_info* pi)
    {
        pi->data = data ;
        pi->size =size ;
        return 0 ;
    }
    
    int process_packet(const packet_info* pi)
    {
        m_last_time = time(0) ;
        send(pi->data,pi->size,0) ;
        return 0 ;
    }
*/
    int get_packet_info(const char* data,int size,packet_info* pi)
    {
        if(memcmp(data+size-2,"\r\n",2)==0)
        {
            pi->data = data ;
            pi->size =size ;
            return 0 ;
        }
        else if ( size > 1500 )
        {
            return -1 ;
        }

        pi->data = data ;
        pi->size = size +1 ;
        return 0 ;
    }
    
    int process_packet(const packet_info* pi)
    {
        static char METHOD[] = "GET" ;
        if(memcmp(pi->data,METHOD,sizeof(METHOD)-1)!=0)  return -1 ;
        
        const char* query_string = pi->data + sizeof(METHOD)-1 ;
        int offset = strspn(query_string," ") ;
        if(offset < 1 ) return -1 ;
        query_string += offset ;
        offset = strcspn(query_string," ") ;
        if(offset < 1 ) return -1 ;
        std::string path(query_string,offset) ;
        //printf("path:%s\n",path.c_str()) ;

        static char OK[] = "HTTP/1.1 200 OK\r\n" ;
        send(OK,sizeof(OK)-1,0) ;
        return -1 ;
    }

    
    void on_connected() ;
    void on_closed(int error_type,int error_no) ;


    void on_timeout(timer_manager* manager) ;

private:
    int m_last_time ;
    
    template_timer<test_connection,&test_connection::on_timeout> m_timer ; 


} ; 

class test_acceptor : public tcp_acceptor
{
public:
    int on_connection(int fd,sa_in_t* addr) ;

private:


};

class event_queue : public io_handler
{
public:
    typedef circular_queue<int> fd_queue ;
public:
    int init(epoll_reactor* reactor)
    {
        if( m_fd_queue.init(1024) < 0) return -1  ;
        m_event_fd = eventfd(0,EFD_NONBLOCK) ;
        if(m_event_fd < 0) return -2 ;
        reactor->add_handler(m_event_fd,this,epoll_reactor::EVENT_READ) ;
        m_reactor = reactor ;
        return 0 ;

    }

    void fini()
    {
        close(m_event_fd) ;
        m_fd_queue.fini() ;
    }

    int push_fd(int fd)
    {
        if(m_fd_queue.push(fd)!=0) return -1 ;
        static int64_t v = 1 ;
        write(m_event_fd,(char*)&v,sizeof(v)) ;
        return 0 ;
    }
    
    ~event_queue() { fini();} ;

protected:
    void on_read(int fd)
    {
        //printf("worker_thread:%lu on_read\n",pthread_self() ) ;
        static char buf[8] ;
        read(m_event_fd,buf,sizeof(buf)) ;
        on_notify() ;
    }

    void on_write(int fd)
    {
    }

    void on_error(int fd)
    {
    }

    void on_notify()
    {
        int fd = 0 ;
        while( m_fd_queue.pop(fd)==0)
        {
            printf("worker_thread:%lu recv fd:%d\n",pthread_self(),fd) ;
            test_connection* conn = new test_connection ;
            if(conn == NULL || conn->init(m_reactor,fd) !=0 )
            {
                close(fd) ;
                continue ;
            }

        }

    }

private:
    epoll_reactor* m_reactor ;
    fd_queue m_fd_queue ;
    int m_event_fd ;
} ;

class worker : public kingnet::simple_thread
{
public:
    virtual ~worker() { } ;

    void set_listen_fd(int listen_fd) 
    {
        m_listen_fd = listen_fd ;
    }

    int push_fd(int fd)
    {
        return m_queue.push_fd(fd) ;
    }


protected:
    int on_init()
    {
        m_reactor.init(1024) ;
        //m_acceptor.init(&m_reactor,m_listen_fd) ;
        m_queue.init(&m_reactor) ;
        printf("worker_thread:%lu on_init\n",pthread_self()) ;
        return 0 ;
    } 

    void on_fini() 
    {
        //m_acceptor.fini() ;
        m_reactor.run_once(1000) ;
        m_queue.fini() ;
        m_reactor.fini() ;
    }

    void run_once() 
    {
        m_reactor.run_once(10000) ;
    }

private:
    epoll_reactor m_reactor ;
    event_queue m_queue ;
    int m_listen_fd ;



} ;

class test_app : public application
{
public:
    enum
    {
        MAX_WORKER_COUNT = 4 ,
    } ;

    int on_init()
    {
        m_acceptor.init(&reactor(),"0.0.0.0",9999) ;
        
        for(int i = 0 ; i < MAX_WORKER_COUNT ; ++i)
        {
            m_workers[i].set_listen_fd(m_acceptor.fd() ) ;
            m_workers[i].start() ;
        }

        printf("main_thread:%lu on_init\n",pthread_self()) ;
        set_timer(10000) ;

        return 0 ;
    }

    void on_fini()
    {
        
        for(int i = 0 ; i < MAX_WORKER_COUNT ; ++i)
        {
            m_workers[i].stop() ;
        }

        m_acceptor.fini() ;
    }

    void on_timer()
    {
        printf("app timer\n") ;
    }

    int dispatch_client(int fd)
    {
        return m_workers[rand()%MAX_WORKER_COUNT].push_fd(fd) ;
    }

private:
    test_acceptor m_acceptor ;
    worker m_workers[MAX_WORKER_COUNT] ;


} ;

void test_connection::on_connected()
{
    m_last_time = time(0) ;
    //printf("connection::on_connected\n") ;
}


void test_connection::on_closed(int error_type,int error_no)
{
    //printf("disconnect type:%d\n",error_type) ;
    delete this ;
}

void test_connection::on_timeout(timer_manager* manager)
{
        int now = time(0) ;
        if(now - m_last_time > 10)
        {
            printf("connection timer\n");
            fini() ;
            return ;
        }


}




int test_acceptor::on_connection(int fd,sa_in_t* addr)
{
    /*
    printf("thread:%lu on_connection\n",pthread_self()) ;
    test_connection* conn = new test_connection ;
    if(conn == NULL ) return -1 ;
    if( conn->init(m_reactor,fd)!=0) return -1 ;
    set_socket_nodelay(conn->get_id().fd) ;
    */

    printf("thread:%lu on_connection\n",pthread_self()) ;
    if(get_app().dispatch_client(fd)!=0) close(fd) ;
    
    return 0 ;
}


IMPLEMENT_APPLICATION_INSTANCE(test_app) ;
IMPLEMENT_MAIN() ;
